"""
@author: xxw
@file: python_consumer.py
@time: 2023/8/13 17:51
"""
import requests
from datetime import datetime
import json
import pymysql

# 将日志时间转换为所需格式
def time_form(raw_time):
    # 解析后的时间格式
    parsed_time = datetime.strptime(raw_time,  "%d/%b/%Y:%H:%M:%S")
    formatted_time = parsed_time.strftime("%Y-%m-%d %H:%M:%S")
    return formatted_time

# 查询ip地址的信息，省份和运营商
def get_location(ip):
    url = f'http://ip-api.com/json/{ip}'
    response = requests.get(url)
    if response.status_code == 200:
        data = response.json()
        prov = data['regionName']
        isp = data['isp']
        return prov, isp
    return None, None

from pykafka import KafkaClient
client = KafkaClient(hosts="192.168.116.129:9092,192.168.116.135:9092,192.168.116.136:9092")
topic = client.topics['nginxlog']
balanced_consumer = topic.get_balanced_consumer(
    consumer_group='testgroup',
    # 自动提交offset
    auto_commit_enable=True,
    zookeeper_connect='nginx-kafka01:2181,nginx-kafka02:2181,nginx-kafka03:2181'
)
db = pymysql.connect(host="192.168.116.129", user="xxw", password="123456", port=3306, db="consumers", charset="utf8")
cursor = db.cursor()
for message in balanced_consumer:
    if message is not None:
        try:
            line = json.loads(message.value.decode("utf-8"))
            log = line["message"]
            tmp_lst = log.split()
            ip = tmp_lst[0]
            raw_time = tmp_lst[3].replace("[", "")
            bt = tmp_lst[9]
            formatted_time = time_form(raw_time)
            prov, isp = get_location(ip)
            if prov and isp:
                print(prov, isp, formatted_time, bt)
                try:
                    sql_statement = f'insert into consumers values("{formatted_time}","{prov}","{isp}","{bt}")'
                    cursor.execute(sql_statement)
                    db.commit()
                    print("保存成功")
                except Exception as e:
                    print("修改失败", e)
                    db.rollback()
        except:
            pass
db.close()




